package com.example.teemor.es_demo.service;

import com.example.teemor.es_demo.config.LogManagementProperties;
import com.example.teemor.es_demo.entity.ServiceLog;
import com.example.teemor.es_demo.entity.UserBehavior;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 日志生产者服务
 * 负责将日志数据批量发送到Kafka
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class LogProducerService {
    
    private final KafkaTemplate<String, String> kafkaTemplate;
    private final LogManagementProperties properties;
    private final ObjectMapper objectMapper;
    
    // 服务日志缓冲队列
    private final ConcurrentLinkedQueue<ServiceLog> serviceLogBuffer = new ConcurrentLinkedQueue<>();
    private final AtomicInteger serviceLogCount = new AtomicInteger(0);
    
    // 用户行为日志缓冲队列
    private final ConcurrentLinkedQueue<UserBehavior> userBehaviorBuffer = new ConcurrentLinkedQueue<>();
    private final AtomicInteger userBehaviorCount = new AtomicInteger(0);
    
    // 定时任务执行器
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
    
    /**
     * 初始化定时任务
     */
    public void init() {
        // 定时发送服务日志
        scheduler.scheduleAtFixedRate(
            this::flushServiceLogs,
            properties.getKafka().getBatch().getServiceLogTimeout(),
            properties.getKafka().getBatch().getServiceLogTimeout(),
            TimeUnit.MILLISECONDS
        );
        
        // 定时发送用户行为日志
        scheduler.scheduleAtFixedRate(
            this::flushUserBehaviorLogs,
            properties.getKafka().getBatch().getUserBehaviorTimeout(),
            properties.getKafka().getBatch().getUserBehaviorTimeout(),
            TimeUnit.MILLISECONDS
        );
        
        log.info("日志生产者服务初始化完成");
    }
    
    /**
     * 发送服务日志
     */
    @Async
    public void sendServiceLog(ServiceLog serviceLog) {
        serviceLogBuffer.offer(serviceLog);
        
        // 检查是否达到批处理大小
        if (serviceLogCount.incrementAndGet() >= properties.getKafka().getBatch().getServiceLogSize()) {
            flushServiceLogs();
        }
    }
    
    /**
     * 发送用户行为日志
     */
    @Async
    public void sendUserBehaviorLog(UserBehavior userBehavior) {
        userBehaviorBuffer.offer(userBehavior);
        
        // 检查是否达到批处理大小
        if (userBehaviorCount.incrementAndGet() >= properties.getKafka().getBatch().getUserBehaviorSize()) {
            flushUserBehaviorLogs();
        }
    }
    
    /**
     * 刷新服务日志缓冲区
     */
    private void flushServiceLogs() {
        if (serviceLogBuffer.isEmpty()) {
            return;
        }
        
        List<ServiceLog> logs = new ArrayList<>();
        ServiceLog log;
        while ((log = serviceLogBuffer.poll()) != null) {
            logs.add(log);
        }
        
        if (!logs.isEmpty()) {
            sendServiceLogBatch(logs);
            serviceLogCount.set(0);
        }
    }
    
    /**
     * 刷新用户行为日志缓冲区
     */
    private void flushUserBehaviorLogs() {
        if (userBehaviorBuffer.isEmpty()) {
            return;
        }
        
        List<UserBehavior> behaviors = new ArrayList<>();
        UserBehavior behavior;
        while ((behavior = userBehaviorBuffer.poll()) != null) {
            behaviors.add(behavior);
        }
        
        if (!behaviors.isEmpty()) {
            sendUserBehaviorBatch(behaviors);
            userBehaviorCount.set(0);
        }
    }
    
    /**
     * 批量发送服务日志到Kafka
     */
    private void sendServiceLogBatch(List<ServiceLog> logs) {
        try {
            String topic = properties.getKafka().getTopics().getServiceLog();
            String message = objectMapper.writeValueAsString(logs);
            
            CompletableFuture<SendResult<String, String>> future = 
                kafkaTemplate.send(topic, message);
            
            future.whenComplete((result, ex) -> {
                if (ex == null) {
                    log.debug("服务日志批量发送成功，数量: {}, Topic: {}", logs.size(), topic);
                } else {
                    log.error("服务日志批量发送失败，数量: {}, Topic: {}", logs.size(), topic, ex);
                }
            });
            
        } catch (JsonProcessingException e) {
            log.error("服务日志序列化失败", e);
        }
    }
    
    /**
     * 批量发送用户行为日志到Kafka
     */
    private void sendUserBehaviorBatch(List<UserBehavior> behaviors) {
        try {
            String topic = properties.getKafka().getTopics().getUserBehavior();
            String message = objectMapper.writeValueAsString(behaviors);
            
            CompletableFuture<SendResult<String, String>> future = 
                kafkaTemplate.send(topic, message);
            
            future.whenComplete((result, ex) -> {
                if (ex == null) {
                    log.debug("用户行为日志批量发送成功，数量: {}, Topic: {}", behaviors.size(), topic);
                } else {
                    log.error("用户行为日志批量发送失败，数量: {}, Topic: {}", behaviors.size(), topic, ex);
                }
            });
            
        } catch (JsonProcessingException e) {
            log.error("用户行为日志序列化失败", e);
        }
    }
    
    /**
     * 销毁资源
     */
    public void destroy() {
        // 发送剩余的日志
        flushServiceLogs();
        flushUserBehaviorLogs();
        
        // 关闭定时任务
        scheduler.shutdown();
        try {
            if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) {
                scheduler.shutdownNow();
            }
        } catch (InterruptedException e) {
            scheduler.shutdownNow();
            Thread.currentThread().interrupt();
        }
        
        log.info("日志生产者服务已销毁");
    }
}